package bing.search.es;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsNodeResponse;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsRequest;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import bing.search.common.ClusterNode;
import bing.search.common.Document;
import bing.search.common.Index;
import bing.search.common.IndexAlreadyExistException;
import bing.search.common.IndexTypeField;

public class EsUtils {

	private static Client client = null;

	/* 获取ES client */
	public static Client getClient() {
		if (client != null) {
			return client;
		}

		Settings settings = Settings.builder().put("cluster.name", "my-application").build();

		try {
			client = new PreBuiltTransportClient(settings)
					.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("lbh3"), 9300))
					.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("lbh4"), 9300))
					.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("lbh5"), 9300));
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return client;
	}

	public static boolean createIndex(String indexName, String shardsNum, String replicasNum) throws IOException {
		// 检查指定的index是否存在
		IndicesExistsResponse indexExist = getClient().admin().indices().prepareExists(indexName).get();
		if (indexExist.isExists()) {
			return false;
		}
		// 创建指定的index
		XContentBuilder settings = XContentFactory.jsonBuilder()
				                   .startObject().field("number_of_shards", shardsNum).field("number_of_replicas", replicasNum).endObject();
		CreateIndexRequestBuilder createIndexRequestBuilder = getClient().admin().indices().prepareCreate(indexName)
				.setSettings(settings);
		CreateIndexResponse createIndexResponse = createIndexRequestBuilder.execute().actionGet();
		if (createIndexResponse.isAcknowledged()) {
			return true;
		}
		return false;
	}

    public static boolean createIndexType(String indexName, String typeName, List<IndexTypeField> settings) throws IOException {
		
    	//构建mapping
    	XContentBuilder mapping=XContentFactory.jsonBuilder().startObject()
				                .startObject(typeName).startObject("properties");
		
		for(IndexTypeField typeField : settings) {
			mapping.startObject(typeField.getFieldName());
			for(Entry<String, String> entry : typeField.getSettings().entrySet()) {
				mapping.field(entry.getKey(), entry.getValue());
			}
			mapping.endObject();
		}
		mapping.endObject().endObject().endObject();
		
		PutMappingRequest putMappingRequest =Requests.putMappingRequest(indexName).type(typeName).source(mapping);
		PutMappingResponse putMappingResponse=getClient().admin().indices().putMapping(putMappingRequest).actionGet();
		
		if(putMappingResponse.isAcknowledged()) {
			return true;
		}
		return false;
	}

	public static boolean createDocumentIndex(String indexName, String typeName) throws Exception {
		// 构建索引
		XContentBuilder settings = XContentFactory.jsonBuilder().startObject()
				// .startObject("settings")
				.field("number_of_shards", 3)// 设置分片的数量
				.field("number_of_replicas", 2)// 设置副本的数量
				// .endObject()
				.endObject();
		// 构建类型映射
		XContentBuilder mapping = XContentFactory.jsonBuilder()
				.startObject().startObject(typeName)
				      .startObject("properties")
				        .startObject("id")
				          .field("type", "long")
				          // .field("store", "no")//设置副本的数量
				        .endObject()
				        .startObject("title")
				          .field("type", "string")
				          .field("analyzer", "ik_smart")
				        .endObject()
				        .startObject("author")
				          .field("type", "string")
				          .field("analyzer", "ik_smart")
				        .endObject()
				        .startObject("describe")
				          .field("type", "string")
				          .field("analyzer", "ik_smart")
				        .endObject()
				        .startObject("content")
				          .field("type", "string")
				          .field("analyzer", "ik_smart")
				        .endObject()
				      .endObject()
				.endObject().endObject();

		IndicesExistsResponse exist = getClient().admin().indices().prepareExists(indexName).get();
		if (exist.isExists()) {
			throw new IndexAlreadyExistException(indexName + " already exists.");
		}
		CreateIndexRequestBuilder createIndexBuilder = getClient().admin().indices().prepareCreate(indexName)
				.setSettings(settings);
		CreateIndexResponse createIndexResponse = createIndexBuilder.execute().actionGet();

		PutMappingRequest mappingRequest = Requests.putMappingRequest(indexName).type(typeName).source(mapping);
		PutMappingResponse createMappingResponse = getClient().admin().indices().putMapping(mappingRequest).actionGet();
		if (createIndexResponse.isAcknowledged() && createMappingResponse.isAcknowledged()) {
			System.out.println(indexName + " index created.");
			return true;
		} else {
			System.out.println(indexName + " index creation failed.");
			return false;
		}
	}

	/**
	 * 获取所有的索引
	 * @return
	 */
	public static List<String> loadAllIndices() {
		GetIndexRequest getIndexRequest=new GetIndexRequest();
		ActionFuture<GetIndexResponse> response =getClient().admin().indices().getIndex(getIndexRequest);
		
		String[] indicesArray=response.actionGet().getIndices();
		return Lists.newArrayList(indicesArray);
	}

	public static String loadIndexByName(String indexName) throws IOException{
		GetIndexRequestBuilder getIndexRequestBuilder =getClient().admin().indices().prepareGetIndex().addIndices(indexName);
		GetIndexResponse response =getIndexRequestBuilder.execute().actionGet();
		
		ImmutableOpenMap<String, List<AliasMetaData>> aliasMap = response.aliases();
		ImmutableOpenMap<String, Settings> settingsMap = response.getSettings();
		ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappingMap = response.getMappings();
		
		ObjectMapper mapper=new ObjectMapper();
		StringBuilder strBuilder = new StringBuilder();
		strBuilder.append("\"aliases\":{");
		if(!aliasMap.isEmpty()) {
			strBuilder.append(mapper.writeValueAsString(aliasMap.get(indexName))).append("},");
		}
		strBuilder.append("},");
		
		strBuilder.append("\"settings\":").append(mapper.writeValueAsString(settingsMap.get(indexName).getAsStructuredMap())).append("},");
		strBuilder.append("\"mappings\":");
		
		ImmutableOpenMap<String, MappingMetaData> typeMap = mappingMap.get(indexName);
		if(!typeMap.isEmpty()) {
			Iterator<MappingMetaData> mappingKeys = typeMap.valuesIt();
			while(mappingKeys.hasNext()) {
				strBuilder.append(mappingKeys.next().source());
			}
		}

		return strBuilder.toString();
	}
	
	/** 
	 * 根据索引名字检索
	 * @param indexName
	 * @return
	 */
    public static Map<String,Object> searchIndexByName(String indexName, int start, int size){
    	GetIndexRequest getIndexRequest=new GetIndexRequest();
    	if(!StringUtils.isEmpty(indexName)) {
    		getIndexRequest.indices(indexName);
    	}
    	ActionFuture<GetIndexResponse> response = getClient().admin().indices().getIndex(getIndexRequest);
    	List<Index> indices = Lists.newArrayList();
    	Map<String,Object> result = Maps.newHashMap();
    	String[] nameArray = response.actionGet().getIndices();
    	
    	if(nameArray.length > start) {
    		for(int i=start; i<nameArray.length && i<(start + size); i++) {
        		Index index = new Index();
        		index.setName(nameArray[i]);
        		index.setVersion("1.0");
        		indices.add(index);
        	}
    	}
    	
    	result.put("dataList", indices);
    	result.put("count", nameArray.length);
		return result;
	}
    
    public static boolean deleteIndexByName(String indexName){
    	DeleteIndexRequest deleteIndexRequest =new DeleteIndexRequest(indexName);
    	ActionFuture<DeleteIndexResponse> actionFuture = getClient().admin().indices().delete(deleteIndexRequest);
    	DeleteIndexResponse response=actionFuture.actionGet();
    	
    	if(response.isAcknowledged()) {
    		return true;
    	}
    	return false;
	}
    
	/**
	 * 创建索引， 对id，title，describe，author四项进行索引
	 */
	public static String addIndex(String index, String type, Document doc) {
		HashMap<String, Object> hashMap = Maps.newHashMap();
		hashMap.put("id", doc.getId());
		hashMap.put("title", doc.getTitle());
		hashMap.put("describe", doc.getDescribe());
		hashMap.put("author", doc.getAuthor());

		IndexResponse response = getClient().prepareIndex(index, type).setSource(hashMap).execute().actionGet();
		return response.getId();
	}

	public static Map<String, Object> search(String searchKey, String index, String type, int start, int size) {
		SearchRequestBuilder builder = getClient().prepareSearch(index);
		builder.setTypes(type);
		builder.setFrom(start);
		builder.setSize(size);

		// 设置高亮字段名称，设置高亮前缀，设置高亮后缀
		HighlightBuilder highlightBuilder = new HighlightBuilder().field("title").field("describe")
				.preTags("<font color='red'>").postTags("</font>");
		builder.highlighter(highlightBuilder);
		// 设置高亮前缀
		// builder.highlighter(new HighlightBuilder().preTags("<font color='red'>"));
		// 设置高亮后缀
		// builder.highlighter(new HighlightBuilder().postTags("</font>"));

		builder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
		if (StringUtils.isNotBlank(searchKey)) {
			builder.setQuery(QueryBuilders.termQuery("title", searchKey));
			builder.setQuery(QueryBuilders.multiMatchQuery(searchKey, "title", "describe"));
		}
		builder.setExplain(true);
		// System.out.println("seach:"+builder.toString());

		SearchResponse searchResponse = builder.get();
		SearchHits hits = searchResponse.getHits();

		long total = hits.getTotalHits();
		Map<String, Object> map = Maps.newHashMap();
		SearchHit[] hitArray = hits.getHits();
		map.put("count", total);

		// List<Map<String,Object>> list=Lists.newArrayList();
		List<Document> list2 = Lists.newArrayList();
		for (SearchHit searchHit : hitArray) {
			Map<String, HighlightField> highlightFields = searchHit.getHighlightFields();
			HighlightField highlightField = highlightFields.get("title");
			Map<String, Object> sources = searchHit.getSource();

			Document doc = new Document();
			if (highlightField != null) {
				Text[] fragments = highlightField.fragments();
				String name = "";
				for (Text text : fragments) {
					name += text;
				}
				// sources.put("title", name);
				doc.setTitle(name);
			} else {
				doc.setTitle(sources.get("title").toString());
			}

			highlightField = highlightFields.get("describe");
			if (highlightField != null) {
				Text[] fragments = highlightField.fragments();
				String describe = "";
				for (Text text : fragments) {
					describe += text;
				}
				// sources.put("describe", describe);
				doc.setDescribe(describe);
			} else {
				doc.setDescribe(sources.get("describe").toString());
			}
			// list.add(sources);
			list2.add(doc);
		}
		map.put("dataList", list2);
		return map;
	}

	public static Map<String, Object> loadClusterInfo() {
		Map<String, Object> clusterInfo = Maps.newHashMap();
		
		ClusterHealthRequest clusterHealthRequest = new ClusterHealthRequest();
		ActionFuture<ClusterHealthResponse> clusterHealthResponse=getClient().admin().cluster().health(clusterHealthRequest);
		ClusterHealthResponse healthResponse = clusterHealthResponse.actionGet();
		
		ClusterStatsRequest clusterStatsRequest = new ClusterStatsRequest();
		ActionFuture<ClusterStatsResponse> clusterStatsResponse =getClient().admin().cluster().clusterStats(clusterStatsRequest);
		ClusterStatsResponse statsResponse=clusterStatsResponse.actionGet();
		
		clusterInfo.put("clusterName", statsResponse.getClusterName().value());
		clusterInfo.put("clusterHealthStatus", statsResponse.getStatus().name());
		
		List<ClusterNode> clusterNodeList=Lists.newArrayList();
		for(ClusterStatsNodeResponse response : statsResponse.getNodes()) {
			ClusterNode node = new ClusterNode();
			node.setName(response.getNode().getName());
			
			node.setRole(response.getNode().getRoles().toString());
			node.setHostName(response.getNode().getHostName());
			node.setAddress(response.getNode().getHostAddress());
			clusterNodeList.add(node);
		}
		clusterInfo.put("clusterNodes", clusterNodeList);
		
		return clusterInfo;
	}
}
